package net.bwie.realtime.jtp.dws.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import jdk.internal.org.objectweb.asm.tree.analysis.Value;
import net.bwie.realtime.jtp.dws.log.bean.PageViewBean;
import net.bwie.realtime.jtp.utils.DateTimeUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: net.bwie.realtime.jtp.dws.log.function
 * @Author: zhangleqing
 * @CreateTime: 2025-08-18  16:09
 * @Description: TODO
 * @Version: 1.0
 */
public class PageViewBeanMapFunction extends RichMapFunction<String, PageViewBean> {

    // 声明变量
    private transient ValueState<String> lastVisitDateState = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 获取状态
        lastVisitDateState = getRuntimeContext().getState(
                new ValueStateDescriptor<String>("last-visit-date", String.class)
        );
    }

    @Override
    public PageViewBean map(String value) throws Exception {
        // 解析json
        JSONObject jsonObject = JSON.parseObject(value);
        JSONObject object = jsonObject.getJSONObject("common");
        JSONObject pageObject = jsonObject.getJSONObject("page");

        // 依据地区编码获取省份名称
        String arValue = object.getString("ar");
        //String province = areaDic.get(arValue);

        // UV访客数
        Long uvCount = 0L;

        // 获取上一次访问日期
        String lastVisitDate = lastVisitDateState.value();
        // 获取当前数据日期
        String visitDate = DateTimeUtil.convertLongToString(jsonObject.getLong("ts"), DateTimeUtil.DATE_FORMAT);

        if (null == lastVisitDate || !lastVisitDate.equals(visitDate)){
            uvCount = 1L;
            lastVisitDateState.update(visitDate);
        }

        // 返回实例对象  前面水位线的开始和结束时间可以写成空
        return new PageViewBean(
                null,
                null,
                object.getString("ba"),
                object.getString("ch"),
                arValue,
                object.getString("is_new"),
                pageObject.get("last_page_id") == null ? 1L : 0L,
                1L,
                pageObject.getLong("during_time"),
                uvCount,
                jsonObject.getLong("ts")
        );
    }

    @Override
    public void close() throws Exception {
        super.close();
    }

}
